package com.linck.im.server.handler;

import com.linck.im.dao.mongo.MongoDBUtil;
import com.linck.im.model.proto.ProtoMsg;
import com.mongodb.client.MongoCollection;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelId;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.bson.Document;

import java.util.concurrent.ConcurrentHashMap;

/**
 * 聊天消息处理器
 *
 * @author: linck
 * @create: 2021-01-26
 */
@Slf4j
public class ServerMessageHandler extends SimpleChannelInboundHandler<ProtoMsg.Message> {

    public static final ConcurrentHashMap<String, ChannelId> session = new ConcurrentHashMap<>();

    public static final ChannelGroup channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ProtoMsg.Message message) throws Exception {
        String to = message.getTo();
        ChannelId channelId = session.get(to);
        Channel channel = null;
        if (channelId != null) {
            channel = channels.find(channelId);
        }
        if (channelId == null || channel == null) {
            MongoCollection<Document> messageCollection = MongoDBUtil.getMessageCollection();
            Document document = new Document();
            document.put("to", message.getTo());
            document.put("from", message.getFrom());
            document.put("time", message.getTime());
            document.put("content", message.getContent());
            messageCollection.insertOne(document);
            log.info("{}用户不在线,消息存入mongodb", to);
        } else {
            log.info("{}用户在线,发送消息", to);
            channel.writeAndFlush(message).addListener(channelFuture -> {
                if (!channelFuture.isSuccess()) {
                    Object unused = channelFuture.get();
                    log.warn("发送消息失败:{}", unused);
                } else {
                    log.info("即时消息发送成功");
                }
            });
        }
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        channels.add(incoming);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Channel incoming = ctx.channel();
        log.info(" channelInactive {} channel关闭", incoming.remoteAddress());
        channels.remove(incoming);
    }
}
